package com.ctjsoft.datacollection.core.execute;

import com.ctjsoft.datacollection.core.log.KettleLogUtil;
import com.ctjsoft.datacollection.core.repository.RepositoryUtil;
import com.ctjsoft.datacollection.entity.KRepository;
import com.ctjsoft.datacollection.entity.KScriptRecord;
import com.ctjsoft.datacollection.mapper.KRepositoryMapper;
import com.ctjsoft.datacollection.service.KScriptRecordService;
import com.ctjsoft.datacollection.util.CollectionUtil;
import com.ctjsoft.datacollection.util.FileUtil;
import com.ctjsoft.datacollection.util.SpringContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.pentaho.di.core.ProgressNullMonitorListener;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.logging.LogLevel;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.repository.AbstractRepository;
import org.pentaho.di.repository.Repository;
import org.pentaho.di.repository.RepositoryDirectoryInterface;
import org.pentaho.di.trans.steps.jobexecutor.JobExecutor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * kettle的kjb脚本执行器
 *
 * @author lyf
 */
@Slf4j
@Component
public class JobExecute {

    static KScriptRecordService scriptRecordService;

    @Autowired
    public KScriptRecordService setKScriptRecordService(KScriptRecordService recordService) {
        if (scriptRecordService == null) {
            scriptRecordService = recordService;
        }
        return scriptRecordService;
    }


    /**
     * kjb执行单元，所有的kjb执行都调用该方法
     *
     * @param rep    kjb所在资源库, 单个文件执行时传入null
     * @param jm     kjb元对象
     * @param params kjb需要的命名参数
     */
    private static String executeJob(Repository rep, JobMeta jm, Map<String, String> params) throws KettleException {

        // JobParam 需要加入到  JobMeta 中进行初始化
        if (CollectionUtil.isNotEmpty(params)) {
            for (Map.Entry<String, String> entry : params.entrySet()) {
                jm.setParameterValue(entry.getKey(), entry.getValue());
            }
        }

        // 通过元数据获取kjb的实例
        Job job = new Job(rep, jm);
        // 开启进程守护
        job.setDaemon(true);
        // 设置日志级别
        job.setLogLevel(jm.getLogLevel());
        // 开始执行kjb
        job.start();
        KScriptRecord one = scriptRecordService.getById(params.get("recordId"));

        one.setLogChannelId(job.getLogChannelId());
        scriptRecordService.saveOrUpdate(one);
        // 线程等待，直到kjb执行完成
        job.waitUntilFinished();
        // 执行完成后获取日志
        String logText = KettleLogUtil.getLogText(job.getLogChannelId(), true, job.getLogDate().getTime());
        // 判断执行过程中是否有错误
        if (job.getErrors() > 0) {
            throw new KettleException(logText);
        }
        return logText;
    }


    /**
     * 运行单个kjb
     *
     * @param fullPathName kjb全路径名
     * @param params       kjb需要的命名参数
     * @param logLevel     日志级别
     */
    public static String run(String fullPathName, Map<String, String> params,
                             LogLevel logLevel) throws KettleException {
        // 通过ktr全路径名获取ktr元数据
        JobMeta jm = new JobMeta(FileUtil.replaceSeparator(fullPathName), null);
        // 设置日志级别
        if (logLevel != null) {
            jm.setLogLevel(logLevel);
        }
        // 开始执行kjb
        return executeJob(null, jm, params);
    }
    /**
     * 获取资源库
     *
     * @param transRepositoryId 资源库id
     * @return {@link AbstractRepository}
     */
    private static AbstractRepository getAbstractRepository(String transRepositoryId) {
        KRepositoryMapper repositoryMapper = SpringContextUtil.getBean(KRepositoryMapper.class);
        KRepository kRepository = repositoryMapper.selectById(transRepositoryId);
        return RepositoryUtil.connection(kRepository);
    }
    /**
     * 运行资源库中的kjb
     *
     * @param repositoryId          资源库对象
     * @param jobPath      kjb所在路径
     * @param jobName      kjb名称
     * @param versionLabel 版本号，传入null表示执行最新的kjb
     * @param params       kjb需要的命名参数
     * @param logLevel     日志级别
     */
    public static String run(String repositoryId, String jobPath, String jobName, String versionLabel,
                             Map<String, String> params, LogLevel logLevel) throws KettleException {

        AbstractRepository repository = getAbstractRepository(repositoryId);
        // 根据相对目录地址获取ktr所在目录信息
        RepositoryDirectoryInterface rdi = repository
                .loadRepositoryDirectoryTree()
                .findDirectory(FileUtil.getParentPath(jobPath));
        // 在指定资源库的目录下找到要执行的转换
        JobMeta jm = repository.loadJob(jobName, rdi, new ProgressNullMonitorListener(), versionLabel);
        // 设置日志级别
        if (logLevel != null) {
            jm.setLogLevel(logLevel);
        }
        // 开始执行kjb
        return executeJob(repository, jm, params);
    }
}
